package com.ydj.controller;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.PostConstruct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.server.ResponseStatusException;

import lombok.extern.slf4j.Slf4j;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
@RequestMapping("/pong")
@Slf4j
public class PongController {
    // 每秒钟产生一个令牌，抢完就会返回429
    public static int maxPingCount = 2;

//    AtomicInteger count = new AtomicInteger(maxPingCount);
//    AtomicInteger count = new AtomicInteger(maxPingCount);
    LinkedHashMap<Integer,AtomicInteger> tokens=new LinkedHashMap<>();
    
    @Value("${tokenResetDuration:5}")
    private Integer tokenResetDuration;
    @Value("${spring.kafka.topic:pingpong}")
    private String topic;
    private Disposable fluxInterval = null;
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @PostConstruct
    public void init() {
        fluxInterval = Flux.interval(Duration.ofSeconds(tokenResetDuration)).subscribe(n -> produceTokens());
    }
    
    private void produceTokens() {
        LocalDateTime now = LocalDateTime.now();
        int curSecond = now.getSecond();
        for(int i=0;i<10;i++) {
            LocalDateTime newDate=now.plusSeconds(i);
            if(tokens.get(curSecond+i)!=null) {
                continue;
            }
            AtomicInteger count = new AtomicInteger(maxPingCount);
            count.set(maxPingCount);
            tokens.put(newDate.getSecond(), count);
        }
        Iterator<Map.Entry<Integer,AtomicInteger>> iterator = tokens.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<Integer,AtomicInteger> entry = iterator.next();
            if (entry.getKey().equals(curSecond)) {
                break;
            }
            iterator.remove(); // 删除元素
        }
        System.out.println("---"+tokens);
        // 小于当前秒的都删掉
//        tokens.entrySet().removeIf(entry->entry.getKey()<curSecond);
    }
    @GetMapping("/say")
    public Mono<String> hello(String say) {
        log.info("recieve msg:{}", say);
        if (tokens.get(LocalDateTime.now().getSecond()).getAndDecrement() > 0) {// && "Hello".equals(say)
            this.sendMessageWithCallback(topic, say);
            log.info("return world");
            return Mono.just("World");
        } else {
            log.info("return 429");
            return Mono.error(new ResponseStatusException(HttpStatus.TOO_MANY_REQUESTS));
        }
    }

    /**
     * set token reset duration
     * 
     * @param duration
     */
    @RequestMapping(value = "/setTokenDuration")
    public void setTokenDuration(Integer duration) {
        fluxInterval.dispose();
        tokenResetDuration = duration;
        fluxInterval = Flux.interval(Duration.ofSeconds(tokenResetDuration)).subscribe(n -> produceTokens());
    }

    private void sendMessageWithCallback(String topic, String message) {
        kafkaTemplate.send(topic, message).addCallback(
                success -> log.info("Sent message with offset: " + success.getRecordMetadata().offset()),
                failure -> {
                    log.error("send error:", failure);
                    throw new RuntimeException("保存到kafka失败！");
                });
    }
}
